package spring.cloud.tasks.client;

import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.springframework.context.ApplicationContext;
import spring.cloud.tasks.client.batch_task_processor.*;
import spring.cloud.tasks.common.utils.ApplicationContextUtils;
import spring.cloud.tasks.common.zookeeper.ZooKeeperBatchTaskDataManager;
import spring.cloud.tasks.common.zookeeper.ZooKeeperDataManager;

import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;

@Slf4j
public class BatchTaskWorker extends Worker {


    @Setter
    @Getter
    private ZooKeeperBatchTaskDataManager zooKeeperBatchTaskDataManager;


    @Setter
    @Getter
    private Map<String, TaskInfo> taskIdToBatchTaskInfoMap = new ConcurrentHashMap<String, TaskInfo>();


    public BatchTaskWorker() {
        try {
            this.ip = InetAddress.getLocalHost().getHostAddress();
        } catch (UnknownHostException e) {
            e.printStackTrace();
        }

        ApplicationContext applicationContext = ApplicationContextUtils.getApplicationContext();
        Map<String, Task> stringTaskDataProcessorMap = applicationContext.getBeansOfType(Task.class);
        for (Task taskDataProcessor : stringTaskDataProcessorMap.values()) {
            taskIdList.add(taskDataProcessor.getTaskId());
        }
    }


    public void initZooKeeperDataManager() throws Exception {
        this.zooKeeperBatchTaskDataManager = new ZooKeeperBatchTaskDataManager(this.zooKeeperManager);
    }

    public void cleanZooKeeperDataManager() {
        this.zooKeeperBatchTaskDataManager = null;
    }

    public ZooKeeperDataManager getZooKeeperDataManager() {
        return zooKeeperBatchTaskDataManager;
    }

    public void stopAll() throws Exception {
        stopBatcherList();
    }

    public void pongpong() throws Exception {
        this.lock.lock();
        try {
            WorkerInfo workerInfo = null;
            boolean fileNotFoundException = false;
            try {
                workerInfo = this.getZooKeeperDataManager().loadWorkerInfo(this.getWorkerId());//总体信息
            } catch (Exception e) {
                log.error("获取workerInfo失败", e);
                fileNotFoundException = true;
            }
            if (fileNotFoundException == true) {//被清理了
                try {
                    stopBatcherList();
                    this.getZooKeeperDataManager().unRegisterWorker(this);//卸载调度器
                } finally {
                    rebuildTaskProcessorListWhenAllowExecuteTasks();
                }
            } else {
                if ("suspend".equals(workerInfo.getMasterManagerStatus())) {
                    stopBatcherList();
                    //暂时不接受新的任务
                } else if ("recovery".equals(workerInfo.getMasterManagerStatus())) {
                    stopBatcherList();
                    rebuildTaskProcessorListWhenAllowExecuteTasks();
                    //
                } else if ("stop".equals(workerInfo.getMasterManagerStatus())) {
                    stopBatcherList();
                    this.stop = true;
                    //清理本地worker然后可以注册新的worker
                } else if ("active".equals(workerInfo.getMasterManagerStatus())) {
                    //正常
                    rebuildTaskProcessorListWhenAllowExecuteTasks();
                } else {
                    //
                }
            }
        } finally {
            this.lock.unlock();
        }
    }


    /**
     * 完全重新构建
     */
    public void stopBatcherList() throws Exception {
        for (String batchTaskId : taskIdToBatchTaskInfoMap.keySet()) {
            TaskInfo taskInfo = taskIdToBatchTaskInfoMap.remove(batchTaskId);
            if (taskInfo != null) {
                taskInfo.stop();
            }
        }
    }


    public void rebuildTaskProcessorListWhenAllowExecuteTasks() throws Exception {
        //只支持worker允许运行的情况下才进行调用
        List<String> taskIdList = this.getZooKeeperDataManager().registerWorkerToPermittedTaskListWhenPongPong(this);
        //这个是允许存活的任务,排开这些之外的需要进行清理
        this.stopTaskProcessorListNotTaskIdList(taskIdList);
        this.rebuildTaskProcessorListByWorkerId(taskIdList);
    }

    private void stopTaskProcessorListNotTaskIdList(List<String> taskIdList) {
        for (String batchTaskId : taskIdToBatchTaskInfoMap.keySet()) {
            if (!taskIdToBatchTaskInfoMap.containsKey(taskIdList)) {
                TaskInfo taskInfo = taskIdToBatchTaskInfoMap.remove(batchTaskId);
                if (taskInfo != null) {
                    taskInfo.stop();
                }
            }
        }
    }

    private void rebuildTaskProcessorListByWorkerId(List<String> taskIdList) throws Exception {
        for (String taskId : taskIdList) {
            if (!taskIdToBatchTaskInfoMap.containsKey(taskId)) {
                TaskInfo taskInfo = new TaskInfo();
                taskInfo.setTaskId(taskId);
                taskIdToBatchTaskInfoMap.put(taskId, taskInfo);
            }
        }
        Date now = new Date();
        long nowLong = now.getTime();
        for (String taskId : taskIdList) {
            //补齐创建对应的处理线程
            TaskInfo taskInfo = taskIdToBatchTaskInfoMap.get(taskId);
            List<String> taskInstanceIdList = zooKeeperBatchTaskDataManager.getTaskInstanceIdList(taskId, workerId);
            //清理多余的实例
            taskInfo.stop(taskInstanceIdList);
            //
            for (String taskInstanceId : taskInstanceIdList) {
                TaskWorkerInstanceContext taskWorkerInstanceContext = taskInfo.taskInstanceIdToWorkerTaskBatchInfoMap.get(taskInstanceId);
                if (taskWorkerInstanceContext == null) {
                    taskWorkerInstanceContext = new TaskWorkerInstanceContext();
                    taskWorkerInstanceContext.setTaskId(taskId);
                    taskWorkerInstanceContext.setTaskInstanceId(taskInstanceId);
                    taskInfo.taskInstanceIdToWorkerTaskBatchInfoMap.put(taskInstanceId, taskWorkerInstanceContext);
                }

                final TaskWorkerInstanceServerInfo taskWorkerInstanceServerInfo = zooKeeperBatchTaskDataManager.getWorkerTaskInstanceBatchInfo(taskId, taskInstanceId);
                if (taskWorkerInstanceServerInfo == null) {
                    taskInfo.stop(taskInstanceId);//有可能临时被删除了
                } else {
                    TaskWorkerInstanceClientInfo taskWorkerInstanceClientInfo = taskWorkerInstanceContext.getTaskWorkerInstanceClientInfo();
                    if (taskWorkerInstanceClientInfo == null) {
                        taskWorkerInstanceClientInfo = new TaskWorkerInstanceClientInfo();
                        taskWorkerInstanceClientInfo.setTaskId(taskId);
                        taskWorkerInstanceClientInfo.setTaskInstanceId(taskInstanceId);
                        taskWorkerInstanceContext.setTaskWorkerInstanceClientInfo(taskWorkerInstanceClientInfo);
                    }
                    //
                    String batchCreaterRun = taskWorkerInstanceServerInfo.getBatchCreaterRun();
                    if ("Y".equals(batchCreaterRun) && taskWorkerInstanceServerInfo.needRunOfBatchCreater() && workerId.equals(taskWorkerInstanceServerInfo.getBatchCreaterRunWorkerId())) {
                        TaskWorkerInstanceClientInfo.TaskThreadInfo taskThreadInfo = taskWorkerInstanceClientInfo.getTaskCreaterThreadInfo();
                        boolean needCreateTaskDealThread = false;
                        if (taskThreadInfo == null) {
                            needCreateTaskDealThread = true;
                        } else {
                            if (now.getTime() - taskThreadInfo.getLastUpdateDate().getTime() > 1000 * 60) {
                                needCreateTaskDealThread = true;
                            }
                        }
                        if (needCreateTaskDealThread) {
                            //设置心跳信息
                            String threadId = TaskCreater.class.getName() + "-" + taskId + "-" + taskInstanceId + "-" + nowLong + "-" + UUID.randomUUID().toString();
                            TaskInstanceExecuterThreadInfo taskInstanceExecuterThreadInfo = new TaskInstanceExecuterThreadInfo();
                            taskInstanceExecuterThreadInfo.setTaskId(taskId);
                            taskInstanceExecuterThreadInfo.setTaskInstanceId(taskInstanceId);
                            taskInstanceExecuterThreadInfo.setThreadId(threadId);
                            taskInstanceExecuterThreadInfo.setThreadType(TaskCreater.class.getName());
                            TaskInstanceExecuterThread thread = new TaskInstanceExecuterThread() {
                                @Override
                                public void run() {
                                    TaskCreater taskCreater = null;
                                    setBatchTaskExecuter(taskCreater);
                                    taskCreater.batchCreate(getBatchTaskWorker(), getTaskInstanceThreadInfo(), getParameter());
                                }
                            };
                            thread.setBatchTaskWorker(this);
                            thread.setTaskInstanceThreadInfo(taskInstanceExecuterThreadInfo);
                            thread.setParameter(taskWorkerInstanceServerInfo.getBatchResultCollecterRunParameter());
                            thread.setDaemon(true);
                            thread.setName(taskId);
                            thread.start();
                            //
                            taskThreadInfo = new TaskWorkerInstanceClientInfo.TaskThreadInfo();
                            taskThreadInfo.setTaskId(taskId);
                            taskThreadInfo.setTaskInstanceId(taskInstanceId);
                            taskThreadInfo.setThreadId(threadId);
                            taskThreadInfo.setLastUpdateDate(now);
                            taskWorkerInstanceClientInfo.setTaskCreaterThreadInfo(taskThreadInfo);
                        }

                    } else {
                        //如之前存在则需要注销
                        taskWorkerInstanceClientInfo.setTaskCreaterThreadInfo(null);
                    }
                    String batchExecuterRun = taskWorkerInstanceServerInfo.getBatchExecuterRun();
                    int threadExecuteCountPerBatcher = taskWorkerInstanceServerInfo.getThreadExecuteCountPerBatcher();
                    if ("Y".equals(batchExecuterRun) && taskWorkerInstanceServerInfo.needRunOfBatchExecuter() && taskWorkerInstanceServerInfo.getBatchCreaterRunWorkerIdList().contains(workerId)) {
                        List<TaskWorkerInstanceClientInfo.TaskThreadInfo> taskThreadInfoList = taskWorkerInstanceClientInfo.getTaskExecuterThreadInfoList();
                        List<TaskWorkerInstanceClientInfo.TaskThreadInfo> taskThreadInfoList1 = new ArrayList<TaskWorkerInstanceClientInfo.TaskThreadInfo>();
                        for (TaskWorkerInstanceClientInfo.TaskThreadInfo taskThreadInfo : taskThreadInfoList) {
                            if (!(now.getTime() - taskThreadInfo.getLastUpdateDate().getTime() > 1000 * 60)) {
                                taskThreadInfoList1.add(taskThreadInfo);
                            }
                        }
                        taskThreadInfoList.clear();
                        taskWorkerInstanceClientInfo.setTaskExecuterThreadInfoList(taskThreadInfoList1);
                        int size = taskThreadInfoList1.size();
                        int taskThreadSize = threadExecuteCountPerBatcher - size;
                        if (taskThreadSize > 0) {
                            //线程弥补
                            for (int i = 1; i <= taskThreadSize; i++) {
                                //设置心跳信息
                                String threadId = TaskExecuter.class.getName() + "-" + taskId + "-" + taskInstanceId + "-" + nowLong + "-" + UUID.randomUUID().toString();
                                TaskInstanceExecuterThreadInfo taskInstanceExecuterThreadInfo = new TaskInstanceExecuterThreadInfo();
                                taskInstanceExecuterThreadInfo.setTaskId(taskId);
                                taskInstanceExecuterThreadInfo.setTaskInstanceId(taskInstanceId);
                                taskInstanceExecuterThreadInfo.setThreadId(threadId);
                                taskInstanceExecuterThreadInfo.setThreadType(TaskExecuter.class.getName());
                                TaskInstanceExecuterThread thread = new TaskInstanceExecuterThread() {
                                    @Override
                                    public void run() {
                                        TaskExecuter taskExecuter = null;
                                        setBatchTaskExecuter(taskExecuter);
                                        taskExecuter.executeBatch(getBatchTaskWorker(), getTaskInstanceThreadInfo(), getParameter());
                                    }
                                };
                                thread.setBatchTaskWorker(this);
                                thread.setTaskInstanceThreadInfo(taskInstanceExecuterThreadInfo);
                                thread.setParameter(taskWorkerInstanceServerInfo.getBatchResultCollecterRunParameter());
                                thread.setDaemon(true);
                                thread.setName(taskId);
                                thread.start();
                                //
                                TaskWorkerInstanceClientInfo.TaskThreadInfo taskThreadInfo = new TaskWorkerInstanceClientInfo.TaskThreadInfo();
                                taskThreadInfo.setTaskId(taskId);
                                taskThreadInfo.setTaskInstanceId(taskInstanceId);
                                taskThreadInfo.setThreadId(threadId);
                                taskThreadInfo.setLastUpdateDate(now);
                                taskWorkerInstanceClientInfo.getTaskExecuterThreadInfoList().add(taskThreadInfo);
                            }
                        } else if (taskThreadSize < 0) {
                            if (threadExecuteCountPerBatcher > 0) {
                                while (taskWorkerInstanceClientInfo.getTaskExecuterThreadInfoList().size() > threadExecuteCountPerBatcher) {
                                    taskWorkerInstanceClientInfo.getTaskExecuterThreadInfoList().remove(taskWorkerInstanceClientInfo.getTaskExecuterThreadInfoList().size() - 1);
                                }
                            }

                        }

                    } else {
                        taskWorkerInstanceClientInfo.getTaskExecuterThreadInfoList().clear();

                    }
                    String batchResultCollecterRun = taskWorkerInstanceServerInfo.getBatchCreaterRun();
                    if ("Y".equals(batchResultCollecterRun) && taskWorkerInstanceServerInfo.needRunOfBatchResultCollecter() && workerId.equals(taskWorkerInstanceServerInfo.getBatchResultCollecterRunWorkerId())) {
                        TaskWorkerInstanceClientInfo.TaskThreadInfo taskThreadInfo = taskWorkerInstanceClientInfo.getTaskResultCollecterThreadInfo();
                        boolean needCreateTaskDealThread = false;
                        if (taskThreadInfo == null) {
                            needCreateTaskDealThread = true;
                        } else {
                            if (now.getTime() - taskThreadInfo.getLastUpdateDate().getTime() > 1000 * 60) {
                                needCreateTaskDealThread = true;
                            }
                        }
                        if (needCreateTaskDealThread) {
                            //设置心跳信息
                            String threadId = TaskResultCollecter.class.getName() + "-" + taskId + "-" + taskInstanceId + "-" + nowLong + "-" + UUID.randomUUID().toString();
                            TaskInstanceExecuterThreadInfo taskInstanceExecuterThreadInfo = new TaskInstanceExecuterThreadInfo();
                            taskInstanceExecuterThreadInfo.setTaskId(taskId);
                            taskInstanceExecuterThreadInfo.setTaskInstanceId(taskInstanceId);
                            taskInstanceExecuterThreadInfo.setThreadId(threadId);
                            taskInstanceExecuterThreadInfo.setThreadType(TaskResultCollecter.class.getName());
                            //
                            TaskInstanceExecuterThread thread = new TaskInstanceExecuterThread() {
                                @Override
                                public void run() {
                                    TaskResultCollecter taskResultCollecter = null;
                                    setBatchTaskExecuter(taskResultCollecter);
                                    taskResultCollecter.batchResultCollecte(getBatchTaskWorker(), getTaskInstanceThreadInfo(), getParameter());
                                }
                            };
                            thread.setBatchTaskWorker(this);
                            thread.setTaskInstanceThreadInfo(taskInstanceExecuterThreadInfo);
                            thread.setParameter(taskWorkerInstanceServerInfo.getBatchResultCollecterRunParameter());
                            thread.setDaemon(true);
                            thread.setName(taskId);
                            thread.start();
                            //
                            taskThreadInfo = new TaskWorkerInstanceClientInfo.TaskThreadInfo();
                            taskThreadInfo.setTaskId(taskId);
                            taskThreadInfo.setTaskInstanceId(taskInstanceId);
                            taskThreadInfo.setThreadId(threadId);
                            taskThreadInfo.setLastUpdateDate(now);
                            //
                            taskWorkerInstanceClientInfo.setTaskResultCollecterThreadInfo(taskThreadInfo);
                        }

                    } else {
                        taskWorkerInstanceClientInfo.setTaskResultCollecterThreadInfo(null);
                    }
                }
            }


        }
    }


}

